package com.clusterswisdom.horn.api.tmall.listener;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clusterswisdom.horn.api.event.EventPool;
import com.clusterswisdom.horn.api.ioc.event.HornEvent;
import com.clusterswisdom.horn.api.tmall.HornTmcClient;
import com.clusterswisdom.horn.api.tmall.TmallInstance;
import com.taobao.api.TaobaoClient;
import com.taobao.api.domain.TmcMessage;
import com.taobao.api.request.TmcMessagesConfirmRequest;
import com.taobao.api.request.TmcMessagesConsumeRequest;
import com.taobao.api.response.TmcMessagesConsumeResponse;

public class TakeEventThread extends Thread {

	private TaobaoClient client;

	private Logger log = LoggerFactory.getLogger(HornTmcClient.class);
	
	public TakeEventThread(TaobaoClient taobaoClient) {
		this.client = taobaoClient;
	}

	public void run() {
		do {
			try {
				long quantity = 100L;
				TmcMessagesConsumeResponse rsp = null;
				do {
					TmcMessagesConsumeRequest req = new TmcMessagesConsumeRequest();
					req.setQuantity(quantity);
					req.setGroupName("default");
					rsp = client.execute(req);
					if (rsp.isSuccess() && rsp.getMessages() != null) {
						for (TmcMessage msg : rsp.getMessages()) {
							log.debug("get a message " + msg.getTopic() + " : " + msg.getContent());
							
							String channel = TmallInstance.getInstance().getConfig().getChannel();
							String[] topics = msg.getTopic().split("[_]");
							JSONObject param = JSON.parseObject(msg.getContent());
							param.put("userNick", msg.getUserNick());
							param.put("id", msg.getId());
							
							HornEvent event = new HornEvent(null, channel, topics[1], topics[2], param);
							EventPool.getPool().executeEvent(event);
							
							// confirm message
							TmcMessagesConfirmRequest cReq = new TmcMessagesConfirmRequest();
							cReq.setGroupName("default");
							cReq.setsMessageIds(String.valueOf(msg.getId()));
							client.execute(cReq);
						}
					}
					
				} while (rsp != null && rsp.isSuccess() && rsp.getMessages() != null && rsp.getMessages().size() == quantity);
				Thread.sleep(3000L);
				
			} catch (Exception e) {
				e.printStackTrace();
				try {
					Thread.sleep(1000L);
					
				} catch (InterruptedException e1) {
					e1.printStackTrace();
					
				}
				
			}
			
		} while (true);
	}
}